package com.zch.rpc.core.net.jetty2;

import com.zch.rpc.core.exception.RpcException;
import com.zch.rpc.core.model.RpcRequestModel;
import com.zch.rpc.core.model.RpcResponseModel;
import com.zch.rpc.core.net.abstracts.Client;
import com.zch.rpc.core.serialize.impl.HessianSerializer;
import com.zch.rpc.core.util.ThrowableUtil;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.BufferingResponseListener;
import org.eclipse.jetty.client.util.BytesContentProvider;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: zch
 * @Date: 2018/12/28 13:50
 * @Description:
 */
public class JettyClient implements Client{
    private Logger logger= LoggerFactory.getLogger(JettyClient.class);
    public void asyncSend(String address, RpcRequestModel rpcRequestModel) throws Exception {
        // do invoke
        postRequestAsync(address, rpcRequestModel);
    }

    /**
     * post request (async)
     *
     * @param address
     * @return
     * @throws Exception
     */
    private void postRequestAsync(String address, RpcRequestModel rpcRequestModel) throws Exception {

        // reqURL
        String reqURL = address;
        if (!address.toLowerCase().startsWith("http")) {
            reqURL = "http://" + address;	// IP:PORT, need parse to url
        }

        // serialize request
        HessianSerializer hessianSerializer=new HessianSerializer();
        byte[] requestBytes = hessianSerializer.serialize(rpcRequestModel);

        // httpclient
        HttpClient httpClient = getJettyHttpClient();

        // request
        Request request = httpClient.newRequest(reqURL);
        request.method(HttpMethod.POST);
        request.timeout(  5000, TimeUnit.MILLISECONDS);		// async, not need timeout
        request.content(new BytesContentProvider(requestBytes));

        // invoke
        request.send(new BufferingResponseListener() {
            @Override
            public void onComplete(Result result) {

                try {

                    // valid status
                    if (result.isFailed()) {
                        throw new RpcException(result.getResponseFailure());
                    }

                    // valid HttpStatus
                    if (result.getResponse().getStatus() != HttpStatus.OK_200) {
                        throw new RpcException("rpc remoting request fail, http HttpStatus["+ result.getResponse().getStatus() +"] invalid.");
                    }

                    // valid response bytes
                    byte[] responseBytes = getContent();
                    if (responseBytes == null || responseBytes.length==0) {
                        throw new RpcException("rpc remoting request fail, response bytes is empty.");
                    }

                    // deserialize response
                    RpcResponseModel responseModel = (RpcResponseModel) hessianSerializer.deserialize(responseBytes, RpcResponseModel.class);

                    // notify response
                   // xxlRpcReferenceBean.getInvokerFactory().notifyInvokerFuture(xxlRpcResponse.getRequestId(), xxlRpcResponse);

                } catch (Exception e){

                    // fail, request finish, remove request
                    if (result.getRequest().getContent() instanceof BytesContentProvider) {
                        try {
                            BytesContentProvider requestCp = (BytesContentProvider) result.getRequest().getContent();
                            RpcRequestModel requestTmp = (RpcRequestModel) hessianSerializer.deserialize(requestCp.iterator().next().array(), RpcRequestModel.class);

                            // error msg
                            String errorMsg = null;
                            if (e instanceof RpcException) {
                                RpcException rpcException = (RpcException) e;
                                if (rpcException.getCause() != null) {
                                    errorMsg = ThrowableUtil.toString(rpcException.getCause());
                                } else {
                                    errorMsg = rpcException.getMessage();
                                }
                            } else {
                                errorMsg = ThrowableUtil.toString(e);
                            }

                            //  make response
                            RpcResponseModel xxlRpcResponse = new RpcResponseModel();
                            xxlRpcResponse.setRequestId(requestTmp.getRequestId());
                            xxlRpcResponse.setErrorMsg(errorMsg);

                            // notify response
                            //xxlRpcReferenceBean.getInvokerFactory().notifyInvokerFuture(xxlRpcResponse.getRequestId(), xxlRpcResponse);

                        } catch (Exception e2) {
                            logger.info(">>>>>>>>>>> rpc, remoting request error, and callback error: " + e2.getMessage());
                            logger.info(e.getMessage(), e);
                        }
                    } else {
                        logger.info(">>>>>>>>>>> rpc, remoting request error.", e);
                    }

                }


            }
        });
    }

    /**
     * make jetty2 http client
     *
     * @return
     * @throws Exception
     */
    private static HttpClient jettyHttpClient;
    public static HttpClient getJettyHttpClient() throws Exception {

        // get
        if (jettyHttpClient != null) {
            return jettyHttpClient;
        }

        // init jetty2 cilent, avoid repeat init
        synchronized (JettyClient.class) {

            // re-get
            if (jettyHttpClient != null) {
                return jettyHttpClient;
            }


            // init jettp httpclient
            jettyHttpClient = new HttpClient();
            jettyHttpClient.setFollowRedirects(false);	                // avoid redirect-302
            jettyHttpClient.setExecutor(new QueuedThreadPool());		// default maxThreads 200, minThreads 8
            jettyHttpClient.setMaxConnectionsPerDestination(10000);	    // limit conn per desc
            jettyHttpClient.start();						            // start

        }

        return jettyHttpClient;
    }
}
